package com.pangyangyang.elasticsearch.plugin.jdbc.query;

import com.pangyangyang.elasticsearch.plugin.exception.MessageRiverException;
import com.pangyangyang.elasticsearch.plugin.jdbc.JDBCConnectionManager;
import com.pangyangyang.elasticsearch.plugin.jdbc.constant.JDBCHandlerConstants;
import com.pangyangyang.elasticsearch.plugin.river.MessageRiverSettings;
import org.apache.commons.lang.StringUtils;

import java.io.IOException;
import java.sql.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

/**
 * Created by yangwang on 2015/1/20.
 */
public class QueryStatementExecutor {

    private String mysqlUrl;
    private MessageRiverSettings settings;
    private String[] columnNames;
    private String sql;

    private final static String variableRegex = "(\\$[\\w]+)";

    public QueryStatementExecutor(MessageRiverSettings settings) {
        this.mysqlUrl = settings.getString(JDBCHandlerConstants.SETTING_KEY_URL);
        this.settings = settings;
        extractColumnNamesAndStatement();
    }

    public List<String> executeQueryIdOnly(Map<String, String> data){
        Connection connection = JDBCConnectionManager.acquireConnection(mysqlUrl);
        PreparedStatement statement = null;
        ResultSet rs = null;
        try{
            statement = connection.prepareStatement(sql);
            setPreparedStatementParameters(statement, data);
            rs = statement.executeQuery();
            return extractIdList(rs);
        } catch (Exception e){
            throw new MessageRiverException("Execute query statement failed.", e);
        } finally {
            JDBCConnectionManager.closeResultSet(rs);
            JDBCConnectionManager.closeStatement(statement);
            JDBCConnectionManager.releaseConnection(connection);
        }
    }

    public List<QueryDataEntity> executeQuery(Map<String, String> data){
        Connection connection = JDBCConnectionManager.acquireConnection(mysqlUrl);
        PreparedStatement statement = null;
        ResultSet rs = null;
        try{
            statement = connection.prepareStatement(sql);
            setPreparedStatementParameters(statement, data);
            rs = statement.executeQuery();
            return convert2EntityList(rs);
        } catch (Exception e){
            throw new MessageRiverException("Execute query statement failed.", e);
        } finally {
            JDBCConnectionManager.closeResultSet(rs);
            JDBCConnectionManager.closeStatement(statement);
            JDBCConnectionManager.releaseConnection(connection);
        }
    }

    public void start(){
        JDBCConnectionManager.registerConnectionPool(settings);
    }

    public void stop(){
        JDBCConnectionManager.unregisterConnectionPool(mysqlUrl);
    }

    protected void extractColumnNamesAndStatement(){
        String preSql = settings.getString(JDBCHandlerConstants.SETTING_KEY_SQL_TEMPLATE);

        List<String> variables = new ArrayList<>();
        Pattern regex = Pattern.compile(variableRegex);
        Matcher matcher = regex.matcher(preSql);
        for (; matcher.find();){
            variables.add(matcher.group(1));
        }

        columnNames = new String[variables.size()];
        for(int i = 0;  i < variables.size(); i++){
            String variable = variables.get(i);
            columnNames[i] = variable.substring(1);
        }
        sql = preSql.replaceAll(variableRegex, "?");
    }

    protected void setPreparedStatementParameters(PreparedStatement preparedStatement, Map<String, String> data) throws SQLException {
        String[] arguments = new String[this.columnNames.length];
        for (int i = 0; i < arguments.length; i++){
            String argument = data.get(this.columnNames[i]);
            if (argument == null){
                //TODO 处理null的情况
//                preparedStatement.setNull();
            } else {
                preparedStatement.setString(i + 1, argument);
            }
        }
    }

    protected List<QueryDataEntity> convert2EntityList(ResultSet rs) throws SQLException, IOException {
        ResultSetMetaData resultSetMetaData = rs.getMetaData();
        List<QueryDataEntity> entityList = new ArrayList<>();
        for (; rs.next();  ){
            QueryDataEntity entity = new QueryDataEntity();
            for (int col = 1; col <= resultSetMetaData.getColumnCount(); col ++){
                String name = resultSetMetaData.getColumnLabel(col);
                String value = rs.getString(col);
                entity.add(name, value);
            }
            entityList.add(entity);
        }
        return entityList;
    }

    protected List<String> extractIdList(ResultSet rs) throws SQLException{
        List<String> idLists = new ArrayList<>();
        while(rs.next()){
            String id = rs.getString("_id");
            if(StringUtils.isBlank(id)){
                continue;
            }
            idLists.add(id);
        }

        return idLists;
    }

    @Override
    public String toString() {
        return "QueryStatementExecutor{" +
                "sql='" + sql + '\'' +
                ", columnNames=" + Arrays.toString(columnNames) +
                '}';
    }
}
